from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json

import datetime
import json


class DateEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        else:
            return json.JSONEncoder.default(self, obj)



def producer(topic,message):
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    producer = KafkaProducer(
        bootstrap_servers=['192.168.2.139:9092'],
        key_serializer=lambda k: json.dumps(k).encode('utf-8'),
        value_serializer=lambda v: json.dumps(v,cls=DateEncoder).encode('utf-8'))
    # 发送三条消息
    future = producer.send(
            topic,
            message,
           )  # 向分区1发送消息
    producer.flush()
    try:
        future.get(timeout=10) # 监控是否发送成功
    except kafka_errors:  # 发送失败抛出kafka_errors
        traceback.format_exc()

if __name__ == '__main__':
    dic = {'name': 'jack', 'create_time': datetime.datetime(2019, 3, 19, 10, 6, 6)}
    print(json.dumps(dic, cls=DateEncoder))
    producer(topic='dmeo',message=dic)


